package com.heima.kafka.listener;

import com.alibaba.fastjson.JSON;
import com.heima.kafka.controller.User;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * @Description TODO
 * @Author bo.li
 * @Date 2023/3/19 14:34
 * @Version 1.0
 */
@Component
public class OrderMesssageListener {

//    @KafkaListener(topics = "topic-test")
//    public void onMessage(String message){
//        System.out.println("收到消息：" + message);
//    }

    @KafkaListener(topics = "topic-test")
    public void message(ConsumerRecord<String,String> consumerRecord, KafkaConsumer<String,String> consumer){

        System.out.println(JSON.parseObject(consumerRecord.value(), User.class));
        System.out.println("偏移量：" + consumerRecord.offset());

        consumer.commitSync();
    }
}